package timeandwindow;

import com.atguigu.pojo.Event;
import com.atguigu.pojo.UrlViewCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;

public class Flink12_LateDataCorrect {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);

             //为啥开启检查点？失败之后重试，mysql必须开检查点才能写进去
         env.enableCheckpointing(2000L);

         //设置水位线生成周期
         env.getConfig().setAutoWatermarkInterval(500L);


             SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2)) // 1. 延迟水位线的推进 ，
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );


        ds.print("INPUT");
         OutputTag<Event> lateOutputTag = new OutputTag<>("lateOutputTag", Types.POJO(Event.class));


         // 需求: 统计每10秒内每个url的点击次数 ， 将统计的结果写出到Mysql表中
         //窗口：键值分区事件时间滚动窗口
         SingleOutputStreamOperator<UrlViewCount> windowDs = ds
                 .keyBy(event -> event.getUrl())
                 .window(
                         TumblingEventTimeWindows.of(Time.seconds(10))
                 )
                 .allowedLateness(Time.seconds(5)) //窗口等待时间
                 .sideOutputLateData(lateOutputTag)//极端数据放侧输出流
                 .aggregate(
                         new AggregateFunction<Event, UrlViewCount, UrlViewCount>() {
                             @Override
                             public UrlViewCount createAccumulator() {
                                 return new UrlViewCount(null, null, null, 0L);
                             }

                             @Override
                             public UrlViewCount add(Event value, UrlViewCount urlViewCount) {

                                 urlViewCount.setCount(urlViewCount.getCount() + 1);
                                 return urlViewCount;
                             }

                             @Override
                             public UrlViewCount getResult(UrlViewCount urlViewCount) {
                                 return urlViewCount;
                             }

                             @Override
                             public UrlViewCount merge(UrlViewCount urlViewCount, UrlViewCount acc1) {
                                 return null;
                             }
                         },
                         new ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>() {
                             @Override
                             public void process(String key, ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>.Context context, Iterable<UrlViewCount> elements, Collector<UrlViewCount> out) throws Exception {
                                 //获得增量聚合结果
                                 UrlViewCount urlViewCount = elements.iterator().next();

                                 //设置url
                                 urlViewCount.setUrl(key);

                                 urlViewCount.setWindowStart(context.window().getStart());
                                 urlViewCount.setWindowEnd(context.window().getEnd());
                                 //输出
                                 out.collect(urlViewCount);
                             }
                         }
                 );
         //将结果写到mysql中
         SinkFunction<UrlViewCount> jdbcSink = JdbcSink.<UrlViewCount>sink(
                 "replace into url_viewz_count(window_start ,window_end, url ,cnt) values (? ,?,?,?)",
                 new JdbcStatementBuilder<UrlViewCount>() {
                     @Override
                     public void accept(PreparedStatement preparedStatement, UrlViewCount urlViewCount) throws SQLException {
                         //给sql赋值
                         preparedStatement.setLong(1,urlViewCount.getWindowStart());
                         preparedStatement.setLong(2,urlViewCount.getWindowEnd());
                         preparedStatement.setString(3,urlViewCount.getUrl());
                         preparedStatement.setLong(4,urlViewCount.getCount());

                     }
                 },
                 new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                         .withDriverName("com.mysql.cj.jdbc.Driver")
                         .withUrl("jdbc:mysql://hadoop102:3306/test")
                         .withUsername("root")
                         .withPassword("000000")
                         .build()
         );
  //将窗口计算的结果写到mysql表里
         windowDs.addSink(jdbcSink);

         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
}
